// Copyright (C) 2021-2021 wuchengwen. All Rights Reserved.
// Description: 线程池实现源文件
// Author : wuchengwen <wuchw@foxmail.com>
// Date :2021-07-09

#include "threadpool.hpp"
#include <chrono>

using namespace std;

ThreadPool::ThreadPool(uint32_t minNum, uint32_t maxNum) :
    minNum_(minNum), maxNum_(maxNum)
{
    lock_guard<mutex> lock(this->poolMutex_);
    // 创建工作线程
    for (uint32_t idx = 0; idx < minNum_; ++idx) {
        threads_.emplace_back(worker, ref(*this));
        liveNum_++;
    }
    // 创建管理者线程
    manageThread_ = thread(manager, ref(*this));
}

ThreadPool::~ThreadPool()
{
    shutDown_ = true;
    // 唤醒所有线程
    notEmpty_.notify_all();
    manageThread_.join();
    for (auto &thd : threads_) {
        thd.join();
    }
}

// 添加任务
bool ThreadPool::AddTask(function<void(void)> &&func) noexcept
{
    {
        // 如果队列满了，需要等待
        unique_lock<mutex> lock(poolMutex_);
        notFull_.wait(lock, [this]() { return taskQue_.size() < queCap_; });
        // 如果线程已经准备退出了，则不在接受新的任务
        if (shutDown_ == true) {
            return false;
        }
        taskQue_.emplace(func);
    }
    notEmpty_.notify_one();
    return true;
}

void ThreadPool::worker(ThreadPool &pool) noexcept
{
    function<void(void)> task;
    while (1) {
        {
            unique_lock<mutex> lock(pool.poolMutex_);
            // 判断是否有任务，且线程没有被shutDown
            pool.notEmpty_.wait(lock, [&pool] { return (pool.taskQue_.size() != 0 || pool.shutDown_ == true); });

            // 如果线程被shutdown,则退出线程
            if (pool.shutDown_ && pool.taskQue_.empty()) {
                return;
            }

            // 需要销毁线程
            if (pool.exitNum_ > 0) {
                pool.exitNum_--;
                pool.liveNum_--;
                return;
            }
            task = move(pool.taskQue_.front());
            pool.taskQue_.pop();
            pool.notFull_.notify_one();
        }
        {
            lock_guard<mutex> lock(pool.busyMutex_);
            pool.busyNum_++;
        }
        // 从队列中读取任务，并执行
        task();
        {
            lock_guard<mutex> lock(pool.busyMutex_);
            pool.busyNum_--;
        }
    }
}

void ThreadPool::manager(ThreadPool &pool) noexcept
{
    // 如果线程池没有关闭, 就一直检测
    while (!pool.shutDown_ || !pool.taskQue_.empty()) {
        // 每隔5s检测一次
        this_thread::sleep_for(5s);
        // 取出线程池中的任务数和线程数量
        uint32_t queueSize = 0;
        uint32_t liveNum = 0;
        uint32_t busyNum = 0;
        {
            lock_guard<mutex> lock(pool.poolMutex_);
            queueSize = pool.taskQue_.size();
            liveNum = pool.liveNum_;
            busyNum = pool.busyNum_;
        }

        // 创建线程
        const uint32_t NUMBER = 2;
        // 当前任务个数>存活的线程数 && 存活的线程数<最大线程个数
        if (queueSize > liveNum && liveNum < pool.maxNum_) {
            // 线程池加锁
            lock_guard<mutex> lock(pool.poolMutex_);
            uint32_t num = 0;
            for (auto &thd : pool.threads_) {
                if (!thd.joinable()) {
                    thd = thread(worker, ref(pool));
                    pool.liveNum_++;
                    ++num;
                }

                if (num >= NUMBER) {
                    break;
                }
            }

            while (num < NUMBER) {
                pool.threads_.emplace_back(worker, ref(pool));
                pool.liveNum_++;
                ++num;
            }
        }
        // 销毁多余的线程
        // 忙线程*2 < 存活的线程数目 && 存活的线程数 > 最小线程数量
        if (busyNum * 2 < liveNum && liveNum > pool.minNum_) {
            lock_guard<mutex> lock(pool.poolMutex_);
            pool.exitNum_ = NUMBER;
        }
    }
}
